package cn.nn200433.mqtt.core;

import cn.hutool.core.util.StrUtil;
import cn.nn200433.mqtt.config.MqttClientProperties;
import cn.nn200433.mqtt.constants.MqttConstants;
import cn.nn200433.mqtt.service.impl.MqttHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.lang.reflect.Constructor;
import java.util.HashMap;
import java.util.Map;

/**
 * mqtt 客户端
 *
 * @author nn200433
 * @date 2022-08-03 11:27:22
 */
@Slf4j
public class MyMqttClient {

    /**
     * 已发布的主题
     */
    public static final     Map<String, MqttTopic> PUBLISH_TOPIC  = new HashMap<String, MqttTopic>();
    private static final    String                 TCP_SERVER_URL = "tcp://{}:{}";
    private static volatile MqttAsyncClient        client         = null;

    /**
     * 高频mqtt客户机
     *
     * @param mqttConfig         mqtt配置
     * @param options            选项
     * @param mqttHandlerContext mqtt处理程序上下文
     * @return
     * @author nn200433
     */
    public MyMqttClient(MqttClientProperties mqttConfig, MqttConnectOptions options,
                        MqttHandlerContext mqttHandlerContext) {
        try {
            if (null == client) {
                synchronized (MqttClient.class) {
                    if (null == client) {
                        // host为主机名，clientId 即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置 clientId 的保存形式，默认为以内存保存
                        final String mqttServerURI = StrUtil.format(TCP_SERVER_URL, mqttConfig.getIp(), mqttConfig.getPort());
                        client = new MqttAsyncClient(mqttServerURI, mqttConfig.getClientId(), new MemoryPersistence());

                        // region mqtt 回调方法，处理收到消息等...
                        Class<MqttCallbackExtended>       callbackClass = (Class<MqttCallbackExtended>) Class.forName(mqttConfig.getMqttCallbackClass());
                        Constructor<MqttCallbackExtended> constructor   = callbackClass.getConstructor(MqttAsyncClient.class, MqttHandlerContext.class, MqttClientProperties.class);
                        client.setCallback(constructor.newInstance(client, mqttHandlerContext, mqttConfig));
                    }
                }
            }

            if (client.isConnected()) {
                client.disconnect();
            }

            client.connect(options);
        } catch (Exception e) {
            log.error("mqtt客户端异常", e);
        }
    }

    /**
     * 订阅某个主题，qos默认为0
     *
     * @param topicFilter
     */
    public void subscribe(String topicFilter) {
        subscribe(topicFilter, MqttConstants.QOS_0);
    }

    /**
     * 订阅某个主题
     *
     * @param topicFilter 主题
     * @param qos         qos
     * @return
     * @author nn200433
     */
    public void subscribe(String topicFilter, int qos) {
        try {
            client.subscribe(topicFilter, qos);
        } catch (Exception e) {
            log.error("订阅主题 {} 失败...", topicFilter, e);
        }
    }

    /**
     * 退订主题
     *
     * @param topicFilter 主题
     * @author nn200433
     */
    public void unsubscribe(String topicFilter) {
        unsubscribe(new String[]{topicFilter});
    }

    /**
     * 退订一批主题
     *
     * @param topicFilter 主题
     * @author nn200433
     */
    public void unsubscribe(String[] topicFilter) {
        try {
            client.unsubscribe(topicFilter);
        } catch (Exception e) {
            log.error("取消订阅主题 {} 失败...", topicFilter, e);
        }
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param topic 主题
     * @param body  消息体
     * @return
     * @author nn200433
     */
    public void publish(String topic, String body) {
        publish(topic, body, MqttConstants.QOS_0, Boolean.FALSE);
    }

    /**
     * 发布
     *
     * @param topic    主题
     * @param body     消息体
     * @param qos      qos
     * @param retained 是否保留
     * @return
     * @author nn200433
     */
    public void publish(String topic, String body, Integer qos, Boolean retained) {
        try {
            if (null == client || !client.isConnected()) {
                throw new Exception("请连接 MQTT 服务....");
            }
            
            /*
            MqttTopic mqttTopic = PUBLISH_TOPIC.get(topic);
            if (null == mqttTopic) {
                mqttTopic = client.getTopic(topic);
                PUBLISH_TOPIC.put(topic, mqttTopic);
            }
            */

            MqttMessage message = new MqttMessage();
            // 消息质量
            message.setQos(qos);
            // 断开链接是否保存消息，true保存
            message.setRetained(retained);
            message.setPayload(body.getBytes());

            // MqttDeliveryToken token = mqttTopic.publish(message);
            IMqttDeliveryToken token = client.publish(topic, message);
            token.waitForCompletion();

            MsgLogUtil.debug(topic, qos, message);
            log.debug("---> 【MQTT】消息发布结果 = {}", token.isComplete());
        } catch (Exception e) {
            log.error("向主题 {} 发送 {} 失败...", topic, body, e);
        }
    }

}
